package com.hy.study.mapper;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;

/**
 * @author huyong
 * @date 2024 年 02 月 06 日 15:10: 12
 */
public class DistributedCacheMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    private final Map<String, String> pdmap = new HashMap<>();
    private final Text k = new Text();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        // 获取缓存的小表数据
        URI[] cacheFiles = context.getCacheFiles();// 获取缓存文件的数据
        for (URI url : cacheFiles) {
            String filePath = url.getPath();
            BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(filePath), "UTF-8"));
            // 按行读
            String line = "";
            while (StringUtils.isNotBlank(line = reader.readLine())) {
                // 切割数据
                String[] split = line.split("\t");
                String pid = split[0];
                String pname = split[1];
                pdmap.put(pid, pname);
            }
            IOUtils.closeStream(reader);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        // 1001	01	1
        String[] split = line.split("\t");
        String orderId = split[0];
        String pname = pdmap.get(split[1]);
        String amount = split[2];
        String vstr = orderId + "\t" + pname + "\t" + amount;
        k.set(vstr);
        context.write(k, NullWritable.get());
    }
}
